package main

import (
	"fmt"
	"github.com/Shopify/sarama"
)

var addr = "192.168.52.133:9092"

func main() {
	////消费
	//go func() {
	//	consumer_test()
	//}()
	//
	//testProduct()
	//
	//time.Sleep(time.Duration(2) * time.Second)

	mata()
}

func mata() {
	fmt.Printf("metadata test\n")

	config := sarama.NewConfig()
	config.Version = sarama.V2_1_0_0

	client, err := sarama.NewClient([]string{addr}, config)
	if err != nil {
		fmt.Printf("metadata_test try create client err :%s\n", err.Error())
		return
	}

	defer client.Close()

	// get topic set
	topics, err := client.Topics()
	if err != nil {
		fmt.Printf("metadata try get topics err %s\n", err.Error())
		return
	}

	fmt.Printf("topics(%d):\n", len(topics))

	for _, topic := range topics {
		fmt.Println(topic)
	}

	// get broker set
	brokers := client.Brokers()
	fmt.Printf("metadata broker set(%d):\n", len(brokers))
	for _, broker := range brokers {
		fmt.Printf("%s\n", broker.Addr())
	}
}

func testProduct() bool {
	fmt.Printf("producer_test\n")
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.NoResponse
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Version = sarama.V2_1_0_0

	producer, err := sarama.NewAsyncProducer([]string{addr}, config)
	if err != nil {
		fmt.Printf("producer_test create producer error :%s\n", err.Error())
		return true
	}

	defer producer.AsyncClose()

	value := "nihao"

	// send message
	msg := &sarama.ProducerMessage{
		Topic: "yyn_test",
		Key:   sarama.StringEncoder("yyn_test"),
		Value: sarama.ByteEncoder(value),
	}

	// 发送消息
	producer.Input() <- msg

	select {
	case suc := <-producer.Successes():
		fmt.Printf("offset: %d,  timestamp: %s", suc.Offset, suc.Timestamp.String())
	case fail := <-producer.Errors():
		fmt.Printf("err: %s\n", fail.Err.Error())
	}
	return false
}
func consumer_test() {
	fmt.Printf("consumer_test")

	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	config.Version = sarama.V0_11_0_2

	// consumer
	consumer, err := sarama.NewConsumer([]string{addr}, config)
	if err != nil {
		fmt.Printf("consumer_test create consumer error %s\n", err.Error())
		return
	}

	defer consumer.Close()

	partition_consumer, err := consumer.ConsumePartition("yyn_test", 0, sarama.OffsetNewest)
	if err != nil {
		fmt.Printf("consumer try create partition_consumer error %s\n", err.Error())
		return
	}
	defer partition_consumer.Close()

	for {
		select {
		case msg := <-partition_consumer.Messages():
			fmt.Printf("consumer msg offset: %d, partition: %d, timestamp: %s, value: %s\n",
				msg.Offset, msg.Partition, msg.Timestamp.String(), string(msg.Value))
		case err := <-partition_consumer.Errors():
			fmt.Printf("consumer err :%s\n", err.Error())
		}
	}

}
